package com.demo.service;

import com.demo.constants.ZookeeperInfo;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@Service
public class ZkNodeService {
    //参考https://www.cnblogs.com/rocky-fang/p/9030438.html

    CountDownLatch countDownLatch = new CountDownLatch(1);
    Stat stat = new Stat();

    public void getNodeData_sync() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT, new create_watcher());
        System.out.println("state:"+zooKeeper.getState());
        countDownLatch.await();
        String node = "/getNodeData_sync";
        zooKeeper.create(node, "hellodata".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("the data of node:" + new String(zooKeeper.getData(node, true, stat)));
        System.out.println("czxid::"+stat.getCzxid()+",mzxid::" + stat.getMzxid() + ",version::" +  stat.getVersion());
        zooKeeper.close();
    }

    public void getNode_async() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_3, ZookeeperInfo.SESSION_TIMEOUT,
                new create_watcher(){
                    @Override
                    public void process(WatchedEvent event) {
                        if(Watcher.Event.KeeperState.SyncConnected == event.getState()){
                            if(Watcher.Event.EventType.None == event.getType() && null == event.getPath()){
                                countDownLatch.countDown();
                            }else if(Watcher.Event.EventType.NodeChildrenChanged == event.getType()){
                                try {
//                                    System.out.println("get Child:" + zooKeeper.getChildren(event.getPath(), true));
                                    System.out.println("get Child:" + event.getPath());
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                });
        countDownLatch.await();
        String node = "/getNode_async";
        String cnode1 = "/getNode_async/c1";
        String cnode2 = "/getNode_async/c2";
        String cnode3 = "/getNode_async/c3";

        zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zooKeeper.create(cnode1, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        zooKeeper.getChildren(node, true, new getNode_async_callback(), null);
        zooKeeper.create(cnode2, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        zooKeeper.getChildren(node, true, new getNode_async_callback(), null);
        zooKeeper.create(cnode3, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        zooKeeper.getChildren(node, true, new getNode_async_callback(), null);

        zooKeeper.delete(cnode1 , -1);
        zooKeeper.delete(cnode2 , -1);
        zooKeeper.delete(cnode3 , -1);
        zooKeeper.delete(node , -1);
    }
    class getNode_async_callback implements AsyncCallback.Children2Callback {
        @Override
        public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
            System.out.println(rc + "::" + path + "::" + ctx + "::" + children + "::" + stat);
        }
    }

    public void getNode_sync() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT,
                new create_watcher(){
                    @Override
                    public void process(WatchedEvent event) {
                        if(Watcher.Event.KeeperState.SyncConnected == event.getState()){
                            if(Watcher.Event.EventType.None == event.getType() && null == event.getPath()){
                                countDownLatch.countDown();
                            }else if(Watcher.Event.EventType.NodeChildrenChanged == event.getType()){
                                try {
//                                    System.out.println("get Child:" + zooKeeper.getChildren(event.getPath(), true));
                                    System.out.println("get Child:" + event.getPath());
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                });
        countDownLatch.await();
        String node = "/getNode_sync";
        String cnode1 = "/getNode_sync/c1";
        String cnode2 = "/getNode_sync/c2";
        String cnode3 = "/getNode_sync/c3";
        zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zooKeeper.create(cnode1, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println(zooKeeper.getChildren(node, true));
        zooKeeper.create(cnode2, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println(zooKeeper.getChildren(node, true));
        zooKeeper.create(cnode3, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println(zooKeeper.getChildren(node, true));
        zooKeeper.delete(cnode1 , -1);
        zooKeeper.delete(cnode2 , -1);
        zooKeeper.delete(cnode3 , -1);
        zooKeeper.delete(node , -1);
    }

    public void persistent_create_async() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT, new create_watcher());
        System.out.println("state:"+zooKeeper.getState());
        countDownLatch.await();
        String node = "/persistent_sync_create";
        String cnode = node+"/c1";
        int version = -1;

        zooKeeper.create(node, "mydelete".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zooKeeper.create(cnode, "mycnodedelete".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        zooKeeper.delete(node , version, new persistent_create_async_callback(), 1);
        zooKeeper.delete(cnode, version, new persistent_create_async_callback(), 2);
        zooKeeper.delete(node , version, new persistent_create_async_callback(), 3);

        System.out.println("success to delete " + cnode);

        zooKeeper.close();
    }
    class persistent_create_async_callback implements AsyncCallback.VoidCallback {
        @Override
        public void processResult(int rc, String path, Object ctx) {
            System.out.println(rc + "::" + path + "::" +ctx);
        }
    }

    public void persistent_create_sync() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT, new create_watcher());
        System.out.println("state:"+zooKeeper.getState());
        countDownLatch.await();
        String node = "/persistent_sync_create";
        String cnode = node+"/c1";
        int version = -1;

        zooKeeper.create(node, "mydelete".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zooKeeper.create(cnode, "mycnodedelete".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        try {
            zooKeeper.delete(node, version);
        } catch (Exception e) {
            System.out.println("faile to delete path : " + node);
        }
        zooKeeper.delete(cnode, version);
        System.out.println("success to delete " + cnode);
        zooKeeper.delete(node , version);
        zooKeeper.close();
    }

    public void ephemeral_create_sync() throws IOException, InterruptedException, KeeperException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT, new create_watcher());
        System.out.println("state:"+zooKeeper.getState());
        countDownLatch.await();
        String node = "/ephemeral_sync_create";
        String path1 = zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("success create znode:"+ path1);
        String path2 = zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("success create znode:"+ path2);
        zooKeeper.close();
    }

    public void ephemeral_create_async() throws IOException, InterruptedException {
        ZooKeeper zooKeeper = new ZooKeeper(ZookeeperInfo.IP_2, ZookeeperInfo.SESSION_TIMEOUT, new create_watcher());
        System.out.println("state:"+zooKeeper.getState());
        countDownLatch.await();
        String node = "/ephemeral_async_create";
        zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new AsyncCreateCallBack(), "my test text...1");
        zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                new AsyncCreateCallBack(), "my test text...2");
        zooKeeper.create(node, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                new AsyncCreateCallBack(), "my test text...3");
    }
    class AsyncCreateCallBack implements AsyncCallback.StringCallback {
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            System.out.println("rc:"+rc+",path:"+path+",ctx:"+ctx+"name,"+name);
        }
    }

    class create_watcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            // 获取事件的状态
            Event.KeeperState keeperState = event.getState();
            Event.EventType eventType = event.getType();
            // 如果是建立连接
            if (Event.KeeperState.SyncConnected == keeperState) {
                if (Event.EventType.None == eventType) {
                    // 如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
                    System.out.println("zk 建立连接");
                    countDownLatch.countDown();
                }
            }
        }
    }

}
